package com.cn.daimajiangxin.flink.sink;

import org.apache.flink.api.connector.sink2.Sink;
import org.apache.flink.api.connector.sink2.SinkWriter;
import org.apache.flink.api.connector.sink2.WriterInitContext;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.sink.v2.DiscardingSink;

import java.io.IOException;

public class CustomSinkDemo {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        DataStream<String> stream = env.fromData("Custom", "Sink", "Example");

        // 使用自定义Sink
        stream.sinkTo(new CustomSink());

        env.execute("Custom Sink Demo");
    }

    // 自定义Sink实现 - 使用新API
    public static class CustomSink implements Sink<String> {
        private static final long serialVersionUID = 1L;

        public SinkWriter createWriter(Sink.InitContext context) throws IOException {
            return new CustomSink.CustomSinkWriter();
        }
        // SinkWriter负责实际的数据写入逻辑
        private class CustomSinkWriter implements SinkWriter<String> {
            private static final long serialVersionUID = 1L;
            // 初始化资源
            public CustomSinkWriter() {
                // 初始化连接、客户端等资源
                System.out.println("CustomSink initialized");
            }

            // 处理每个元素
            @Override
            public void write(String value, Context context)  throws IOException, InterruptedException {
                // 实际的写入逻辑
                System.out.println("Writing to custom sink: " + value);
            }

            // 刷新缓冲区
            @Override
            public void flush(boolean endOfInput) {
                // 刷新逻辑（如果需要）
            }

            // 清理资源
            @Override
            public void close() throws Exception {
                // 关闭连接、客户端等资源
                System.out.println("CustomSink closed");
            }
        }
    }

}
